package com.shujia.flink.core
import  org.apache.flink.streaming.api.scala._
object Demo01WordCount {
  def main(args: Array[String]): Unit = {

    /**
     * 创建Flink环境
     */
  //获取一个环境，flink在内部会自动判断是本地还是集群
  val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    /**
     * 设置flink程序计算的并行度
     */
    //env.setParallelism(4)


    /**
     * 读取数据
     * 在linux中启动一个socket模拟数据流数据
     * nc -lk 8888
     */

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    /**
     * 设置数据从上游发送到下游的缓存的超时时间
     *
     * 默认超时时间200毫秒
     * 当设置为0时，每一条数据发送一次  ---会降低flink的吞吐量
     *
     */
    env.setBufferTimeout(200)




    /**
     * 统计单词的数量
     */

  //1.将一行转换成多行
  val wordsDS: DataStream[String] = linesDS.flatMap(_.split(","))

    //2.转换成kv格式
    val kvDS: DataStream[(String, Int)] = wordsDS.map((_, 1))


    //3.按照单词分组
    val keyKvDS: KeyedStream[(String, Int), String] = kvDS.keyBy(_._1)

    //4.统计单词的数量，指定对哪一列求和，指定下标，下标从0开始
    val countDS: DataStream[(String, Int)] = keyKvDS.sum(1)


    //打印结果
    countDS.print()

    /**
     * 启动flink
     *
     */
  env.execute()
  }

}
